package com.shujia.flink.sink;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;


import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;


/**
 * @author SHUJIA
 */
public class Demo5MySqlSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> lines = env.socketTextStream("master", 8888);

        /*
         * 实时统计单词的数量，将统计结果保存到mysql
         */
        //一行转换成多行
        DataStream<Tuple2<String, Integer>> kvs = lines
                .flatMap((line, collector) -> {
                    String sep = ",";
                    for (String word : line.split(sep)) {
                        collector.collect(Tuple2.of(word, 1));
                    }
                }, Types.TUPLE(Types.STRING, Types.INT));

        //分组计算单词额数量
        DataStream<Tuple2<String, Integer>> count = kvs.keyBy(kv -> kv.f0).sum(1);

        //将计算结果保存到mysql
        count.addSink(new MySqlSink());

        env.execute();

    }

    /**
     * SinkFunction
     * RichSinkFunction比SinkFunction多了open和close方法，用于初始化和回收资源
     */
    static class MySqlSink extends RichSinkFunction<Tuple2<String, Integer>> {

        Connection con;

        /**
         * open方法在任务启动的时候执行一次，每一个task执行一次，用于创建数据库连接
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            long start = System.currentTimeMillis();
            //1、加载驱动
            Class.forName("com.mysql.jdbc.Driver");
            //2、创建数据库连接
            //创建数据库连接需要的时间太长，
            con = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456");

            long end = System.currentTimeMillis();
            System.out.println(end - start);
        }

        //在任务取消的时候执行，用于回收资源
        @Override
        public void close() throws Exception {
            //6、关闭数据库连接
            con.close();
        }

        /**
         * invoke: 每一条数据执行一次，如果在invoke中创建数据库连接，性能很低
         *
         * @param kv      The input record.
         * @param context Additional context about the input record.
         */
        @Override
        public void invoke(Tuple2<String, Integer> kv, Context context) throws Exception {
            String word = kv.f0;
            Integer num = kv.f1;

            //3、编写sql
            //replace into 替换插入，如果不存在就插入，如果存在就替换，根据主键进行替换，所以需要设置主键
            PreparedStatement stat = con.prepareStatement("replace into clazz_num values(?,?)");
            //4、给字段赋值
            stat.setString(1, word);
            stat.setInt(2, num);
            //5、执行插入
            stat.execute();
        }
    }
}
